package com.atguigu.gmall.realtime.app.dws.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


//模板模式 提取公共维度join功能，把需要实现交给子类
public  abstract class AbsDimAsyncFuntion<T> extends RichAsyncFunction<T,T> implements  DimJoinFuntion<T> {

    //    2.1  一般情况查数据库的访问方式都是同步的
    //    2.2  构建线程池
    //    2.3   把各种请求提交到线程池，进行处理
    //    2.4   如果处理完成，在线程的任务中调用flink提供的
    //resultFuture.complete(Collections.singleton(结果));
    //resultFuture 是回调接口，用于通知Flink该条处理完成，并返回结果。

    public String tableName=null;
    public AbsDimAsyncFuntion(String tableName){
        this.tableName=tableName;
    }


    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {

        //获得连接池
        ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.getPool();

        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                //
                System.out.println("线程："+ Thread.currentThread()+  "异步查询维度....");

                //维度查询
                String id = getJoinId(t);
                JSONObject dimJsoObj= DimUtil.getDimInfoWithCache( tableName, id);

                join(t,dimJsoObj);


                System.out.println("线程："+ Thread.currentThread()+  "异步查询维度完成..");
                resultFuture.complete(Collections.singleton(t));
            }
        });

    }
}
